package com.atguigu.userprofile.app;

import com.atguigu.userprofile.bean.TagInfo;
import com.atguigu.userprofile.constants.ConstCodes;
import com.atguigu.userprofile.dao.TagInfoDAO;
import com.atguigu.userprofile.utils.MyClickhouseUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.sql.SparkSession;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

public class TaskBitmapApp {
    public static void main(String[] args) {
        //为了能使用spark脚本提交任务，需要创建spark环境
        SparkConf sparkConf = new SparkConf().setAppName("TaskBitmapApp");
        SparkSession sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate();

        //0.获取外部传参
        String taskId = args[0];
        String busiDate = args[1];

        //1.获取所有已启用的Tag_info数据
        List<TagInfo> tagInfoWithOn = TagInfoDAO.getTagInfoWithOn();

        //2.创建存放不同标签值类型的tag_code集合
        ArrayList<String> longList = new ArrayList<>();
        ArrayList<String> decimalList = new ArrayList<>();
        ArrayList<String> stringList = new ArrayList<>();
        ArrayList<String> dateList = new ArrayList<>();

        //3.遍历集合那倒tag_value_type，然后通过类型判断要写入的表
        for (TagInfo tagInfo : tagInfoWithOn) {
            if (ConstCodes.TAG_VALUE_TYPE_LONG.equals(tagInfo.getTagValueType())) {
                longList.add(tagInfo.getTagCode());
            } else if (ConstCodes.TAG_VALUE_TYPE_DECIMAL.equals(tagInfo.getTagValueType())) {
                decimalList.add(tagInfo.getTagCode());
            } else if (ConstCodes.TAG_VALUE_TYPE_STRING.equals(tagInfo.getTagValueType())) {
                stringList.add(tagInfo.getTagCode());
            } else if (ConstCodes.TAG_VALUE_TYPE_DATE.equals(tagInfo.getTagValueType())) {
                dateList.add(tagInfo.getTagCode());
            }
        }

        //4.根据不同的类型往不同的表里写数据
        //获取宽表的表名
        String sourceTableName = "up_tag_merge_" + busiDate.replaceAll("-", "");

        insertBitMapSQL("user_tag_value_long", busiDate, longList, sourceTableName);
        insertBitMapSQL("user_tag_value_decimal", busiDate, decimalList, sourceTableName);
        insertBitMapSQL("user_tag_value_string", busiDate, stringList, sourceTableName);
        insertBitMapSQL("user_tag_value_date", busiDate, dateList, sourceTableName);
    }

    //动态拼接插入语句
    public static void insertBitMapSQL(String insertTableName, String busiDate, List<String> tagCodeList, String sourceTableName) {
        if (tagCodeList.size()>0) {
            List<String> list = tagCodeList.stream().map(tagCodeStr -> "('" + tagCodeStr.toLowerCase() + "'," + tagCodeStr.toLowerCase() + ")").collect(Collectors.toList());
            String codeSQL = StringUtils.join(list, ",");


            String insertSQL = "insert into " + insertTableName + " select\n" +
                    "    tag_code,\n" +
                    "    tag_value,\n" +
                    "    groupBitmapState(cast(uid as UInt64)),\n" +
                    "    '" + busiDate + "'\n" +
                    "FROM\n" +
                    "(\n" +
                    "    SELECT\n" +
                    "        tp.1 AS tag_code,\n" +
                    "        tp.2 AS tag_value,\n" +
                    "        uid\n" +
                    "    FROM\n" +
                    "    (\n" +
                    "        SELECT\n" +
                    "            uid,\n" +
                    "            arrayJoin([" + codeSQL + "]) AS tp\n" +
                    "        FROM " + sourceTableName + "\n" +
                    "    )\n" +
                    ")\n" +
                    "GROUP BY\n" +
                    "    tag_code,\n" +
                    "    tag_value";

            //保证数据的一致性
            String dropData = "alter table " + insertTableName + " delete where dt = '" + busiDate + "'";
            MyClickhouseUtil.executeSql(dropData);
            MyClickhouseUtil.executeSql(insertSQL);
        }
    }
}
